﻿//========================================================= 
/**@file lmdb_event.h
 * @brief 使用lmdb进行进程间事件通信的模块
 * 
 * @date 2018-10-12   16:13:54
 * @author Zhyioo 
 * @version 1.0
 */ 
//--------------------------------------------------------- 
#ifndef _LIBZHOUYB_LMDB_EVENT_H_
#define _LIBZHOUYB_LMDB_EVENT_H_
//--------------------------------------------------------- 
#include "lmdb_extractor.h"

#include "../../extension/security/MD5_Provider.h"
using zhou_yb::extension::security::MD5_Provider;
using zhou_yb::extension::security::IHashAlgorithm;

#include <thread>
#include <mutex>
#include <condition_variable>
#include <memory>
#include <vector>

#include <boost/interprocess/sync/named_mutex.hpp>
#include <boost/interprocess/sync/named_semaphore.hpp>
#include <boost/interprocess/exceptions.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
//--------------------------------------------------------- 
namespace zhou_yb {
namespace wrapper {
namespace lmdb_extractor {
//--------------------------------------------------------- 
/// 服务本地端连接名称
#define LOCAL_END "RECV"
/// 远程客户端连接名称
#define REMOTE_END "SEND"
/// 连接退出事件名称
#define DISCONNECT_END "OnExit"
//--------------------------------------------------------- 
/**
 * @brief 使用named_mutex封装的信号
 * @date 2018-10-27 12:38:13
 */
class mutex_signal
{
protected:
    std::shared_ptr<boost::interprocess::named_mutex> _mutex;

    mutex_signal(boost::interprocess::named_mutex* pMutex) : _mutex(pMutex) {}
public:
    static mutex_signal create_only(const char* name)
    {
        boost::interprocess::named_mutex* pMutex = nullptr;
        try
        {
            pMutex = new boost::interprocess::named_mutex(boost::interprocess::create_only, name);
            pMutex->lock();
        }
        catch(const boost::interprocess::interprocess_exception&) {}
        return mutex_signal(pMutex);
    }
    static mutex_signal open_only(const char* name)
    {
        boost::interprocess::named_mutex* pMutex = nullptr;
        try
        {
            pMutex = new boost::interprocess::named_mutex(boost::interprocess::open_only, name);
        }
        catch(const boost::interprocess::interprocess_exception&) {}
        return mutex_signal(pMutex);
    }
    static mutex_signal open_or_create(const char* name)
    {
        mutex_signal mutex = open_only(name);
        if(!mutex.is_valid())
        {
            mutex = create_only(name);
        }
        return mutex;
    }
    static bool remove(const char* name)
    {
        return boost::interprocess::named_mutex::remove(name);
    }
    static int timed_wait(boost::interprocess::named_mutex& mutex, uint timeoutMs, 
        Ref<IInterrupter> interrupter = Ref<IInterrupter>())
    {
        if(timeoutMs < 1)
        {
            mutex.lock();
            return DeviceError::Success;
        }
        Timer timer;
        while(timer.Elapsed() < timeoutMs)
        {
            if(mutex.try_lock()) return DeviceError::Success;
            /* 中断支持 */
            if(!interrupter.IsNull() && InterruptBehavior::Implement(*interrupter))
                return DeviceError::OperatorInterruptErr;
            Timer::Wait(DEV_OPERATOR_INTERVAL);
        }
        return DeviceError::WaitTimeOutErr;
    }

    mutex_signal() {}
    inline bool is_valid() const
    {
        return _mutex != nullptr;
    }
    inline void notify()
    {
        _mutex->unlock();
    }
    inline void wait()
    {
        _mutex->lock();
    }
    inline bool try_wait()
    {
        return _mutex->try_lock();
    }
    bool wait(uint timeoutMs)
    {
        boost::posix_time::ptime timeout = boost::posix_time::second_clock::universal_time();
        timeout += boost::posix_time::milliseconds(timeoutMs);
        return _mutex->timed_lock(timeout);
    }
};
//--------------------------------------------------------- 
/**
 * @brief 使用named_semaphore封装的信号
 * @date 2018-10-27 12:56:27
 */
class semaphore_signal
{
protected:
    std::shared_ptr<boost::interprocess::named_semaphore> _semaphore;

    semaphore_signal(boost::interprocess::named_semaphore* pSemaphore) : _semaphore(pSemaphore) {}
public:
    static semaphore_signal create_only(const char* name)
    {
        boost::interprocess::named_semaphore* pMutex = nullptr;
        try
        {
            pMutex = new boost::interprocess::named_semaphore(boost::interprocess::create_only, name, 0);
        }
        catch(const boost::interprocess::interprocess_exception&) {}
        return semaphore_signal(pMutex);
    }
    static semaphore_signal open_only(const char* name)
    {
        boost::interprocess::named_semaphore* pSemaphore = nullptr;
        try
        {
            pSemaphore = new boost::interprocess::named_semaphore(boost::interprocess::open_only, name);
            while(pSemaphore->try_wait());
        }
        catch(const boost::interprocess::interprocess_exception&) {}
        return semaphore_signal(pSemaphore);
    }
    static semaphore_signal open_or_create(const char* name)
    {
        semaphore_signal semaphore = open_only(name);
        if(!semaphore.is_valid())
        {
            semaphore = create_only(name);
        }
        return semaphore;
    }
    static bool remove(const char* name)
    {
        return boost::interprocess::named_semaphore::remove(name);
    }
    static int timed_wait(boost::interprocess::named_semaphore& semaphore, uint timeoutMs,
        Ref<IInterrupter> interrupter = Ref<IInterrupter>())
    {
        if(timeoutMs < 1)
        {
            semaphore.wait();
            return DeviceError::Success;
        }
        Timer timer;
        while(timer.Elapsed() < timeoutMs)
        {
            if(semaphore.try_wait()) return DeviceError::Success;
            /* 中断支持 */
            if(!interrupter.IsNull() && InterruptBehavior::Implement(*interrupter))
                return DeviceError::OperatorInterruptErr;
            Timer::Wait(DEV_OPERATOR_INTERVAL);
        }
        return DeviceError::WaitTimeOutErr;
    }

    semaphore_signal() {}
    inline bool is_valid() const
    {
        return _semaphore != nullptr;
    }
    inline void notify()
    {
        _semaphore->post();
    }
    inline void wait()
    {
        _semaphore->wait();
    }
    inline bool try_wait()
    {
        return _semaphore->try_wait();
    }
    bool wait(uint timeoutMs)
    {
        boost::posix_time::ptime timeout = boost::posix_time::second_clock::universal_time();
        timeout += boost::posix_time::milliseconds(timeoutMs);
        return _semaphore->timed_wait(timeout);
    }
};
//--------------------------------------------------------- 
/**
 * @brief 带进程间全局命名的实例对象
 * @date 2018-10-28 13:23:00
 */
class named_instance
{
public:
    static string instanceKey(const string& name)
    {
        return name + ".instance";
    }
    static string getHash(const string& key)
    {
        MD5_Provider md5Provider;
        ByteBuilder md5(16);
        IHashAlgorithm::Stream(md5Provider, ByteArray(key.c_str(), key.length()), md5);
        ByteBuilder md5Ascii(32);
        ByteConvert::ToAscii(md5, md5Ascii);
        return string(md5Ascii.GetString(), md5Ascii.GetLength());
    }
    static bool is_exist(const string& name)
    {
        string key = instanceKey(name);
        string id = getHash(key);
        try
        {
            boost::interprocess::named_mutex mutex(boost::interprocess::open_only, id.c_str());
            return true;
        }
        catch(const boost::interprocess::interprocess_exception&) {}
        return false;
    }
    static bool remove(const string& name)
    {
        string id = getHash(instanceKey(name));
        return boost::interprocess::named_mutex::remove(id.c_str());
    }
protected:
    string _instanceName;
    string _instanceId;
    std::shared_ptr<boost::interprocess::named_mutex> _instanceMutex;
public:
    named_instance() {}
    named_instance(const char* name) { create(name); }
    virtual ~named_instance() {}

    bool create(const char* name)
    {
        _instanceName = _strput(name);
        _instanceId = getHash(instanceKey(_instanceName));
        try
        {
            _instanceMutex = std::make_shared<boost::interprocess::named_mutex>(boost::interprocess::create_only, _instanceId.c_str());
            return true;
        }
        catch(const boost::interprocess::interprocess_exception&) { _instanceMutex = nullptr; }
        return false;
    }
    inline const string& getId() const
    {
        return _instanceId;
    }
    inline const string& getName() const
    {
        return _instanceName;
    }
    inline bool is_valid() const
    {
        return _instanceMutex != nullptr;
    }
    bool dispose()
    {
        if(_instanceMutex == nullptr || _instanceMutex.use_count() > 1)
            return false;
        boost::interprocess::named_mutex::remove(_instanceId.c_str());
        _instanceMutex = nullptr;
        return true;
    }
};
//--------------------------------------------------------- 
/**
 * @brief 使用lmdb进行本地RPC通讯的事件封装
 * @date 2018-10-12 17:06:04
 */
template<class T>
class lmdb_signal
{
public:
    typedef T signal_type;
    typedef lmdb_signal<signal_type> this_type;

    /**
     * @brief 删除指定名称的信号
     * @date 2018-11-22 22:40:26
     */
    static bool remove(const string& name)
    {
        string sHash = named_instance::getHash(name);
        signal_type signal = signal_type::open_only(sHash.c_str());
        if(!signal.is_valid())
            return false;
        signal.try_wait();
        signal.notify();
        return signal_type::remove(sHash.c_str());
    }
    /**
     * @brief 带中断控制的超时等待
     * @date 2018-10-14 14:21:36
     */
    template<class TLmdbConverter>
    static int timed_wait(signal_type& signal, const char* db, const string& key,
        typename TLmdbConverter::value_type& data, uint timeoutMs, 
        Ref<IInterrupter> interrupter = Ref<IInterrupter>())
    {
        bool is_notify = false;
        Timer timer;
        while(timer.Elapsed() < timeoutMs)
        {
            if(!is_notify)
            {
                if(signal.wait(DEV_OPERATOR_INTERVAL))
                {
                    is_notify = true;
                    timer.Restart();
                }
            }
            if(is_notify)
            {
                typedef lmdb_cursor_const_iterator<lmdb_string_converter, TLmdbConverter> const_iterator;
                const_iterator itr = lmdb_helper::open_readonly_cursor(db);
                if(itr.get(key, data))
                    return DeviceError::Success;
            }
            /* 中断支持 */
            if(!interrupter.IsNull() && InterruptBehavior::Implement(*interrupter))
                return DeviceError::OperatorInterruptErr;
            Timer::Wait(DEV_OPERATOR_INTERVAL);
        }
        return DeviceError::WaitTimeOutErr;
    }
    template<class TLmdbConverter>
    static bool write(signal_type& signal, lmdb_env& env, const string& key, const typename TLmdbConverter::value_type& data)
    {
        typedef lmdb_cursor_iterator<lmdb_string_converter, TLmdbConverter> iterator;
        iterator itr = lmdb_helper::open(env);
        if(itr.put_commit(key, data))
        {
            signal.notify();
            return true;
        }
        return false;
    }
    template<class TLmdbConverter>
    static bool read(signal_type& signal, lmdb_const_env& env, const string& key, typename TLmdbConverter::value_type& data, uint timeoutMs = 0)
    {
        if(!signal.wait(timeoutMs))
            return false;
        typedef lmdb_cursor_const_iterator<lmdb_string_converter, TLmdbConverter> const_iterator;
        const_iterator itr = open_readonly(env);
        return itr.get(key, data);
    }

    static string signal_key(const string& name, const string& cmd)
    {
        string key = name;
        key += ".";
        key += cmd;
        return named_instance::getHash(key);
    }
    static bool signal_notify(const char* key)
    {
        signal_type signal = signal_type::open_only(key);
        if(!signal.is_valid()) return false;
        signal.notify();
        return true;
    }
    static int signal_wait(signal_type& signal, uint timeoutMs, Ref<IInterrupter> interrupter = Ref<IInterrupter>())
    {
        if(timeoutMs < 1)
        {
            signal.wait();
            return DeviceError::Success;
        }
        Timer timer;
        while(timer.Elapsed() < timeoutMs)
        {
            if(signal.try_wait())
                return DeviceError::Success;
            /* 中断支持 */
            if(!interrupter.IsNull() && InterruptBehavior::Implement(*interrupter))
                return DeviceError::OperatorInterruptErr;
            Timer::Wait(DEV_OPERATOR_INTERVAL);
        }
        return DeviceError::WaitTimeOutErr;
    }
    static int signal_wait(const char* key, uint timeoutMs, Ref<IInterrupter> interrupter = Ref<IInterrupter>())
    {
        Timer timer;
        while(timer.Elapsed() < timeoutMs)
        {
            // 信号可能还未创建,需要在超时时间内重复尝试
            signal_type signal = signal_type::open_or_create(key);
            if(signal.is_valid())
                return signal_wait(signal, timeoutMs, interrupter);
            /* 中断支持 */
            if(!interrupter.IsNull() && InterruptBehavior::Implement(*interrupter))
                return DeviceError::OperatorInterruptErr;
            Timer::Wait(DEV_OPERATOR_INTERVAL);
        }
        return DeviceError::WaitTimeOutErr;
    }
protected:
    string _key;
    signal_type _get(const char* key, bool isOpenOnly)
    {
        return isOpenOnly ? signal_type::open_only(key) : signal_type::open_or_create(key);
    }
public:
    lmdb_signal() {}
    lmdb_signal(const char* key) { create(key); }

    this_type& create(const char* key)
    {
        _key = _strput(key);
        return (*this);
    }
    inline const string& key() const
    {
        return _key;
    }

    signal_type get(bool isOpenOnly = false)
    {
        return _get(_key.c_str(), isOpenOnly);
    }
    signal_type get(const char* cmd, bool isOpenOnly = false)
    {
        string key = _key;
        key += ".";
        key += _strput(cmd);
        string id = named_instance::getHash(key);
        return _get(key.c_str(), isOpenOnly);
    }
};
//--------------------------------------------------------- 
/**
 * @brief lmdb事件RPC通讯连接
 *
 * @description 每个连接维护两个lmdb库,本地端可写(长连接),远端只读(短连接)
 *
 * @date 2018-10-14 11:13:44
 */
template<class T>
class lmdb_event_connection : public named_instance
{
public:
    typedef T signal_type;
protected:
    signal_type _wevent;
    string _sendkey;
    lmdb_env _wenv;

    signal_type _revent;
    string _recvkey;
    string _rdb;
    lmdb_const_env _renv;

    string _createEvent(signal_type& signal, const char* key, bool isOpenOnly = false)
    {
        string id = _instanceId;
        if(!_is_empty_or_null(key))
        {
            id = named_instance::getHash(key);
            key = id.c_str();
        }
        signal = isOpenOnly ? signal_type::open_only(id.c_str()) : signal_type::open_or_create(id.c_str());
        return id;
    }
public:
    lmdb_event_connection() {}
    lmdb_event_connection(const char* name) { create(name); }
    virtual ~lmdb_event_connection() {}

    inline bool is_can_write() const
    {
        return _wevent.is_valid() && _wenv.is_valid();
    }
    inline bool is_can_read() const
    {
        return _revent.is_valid() && (_renv.is_valid() || _rdb.length() > 0);
    }
    /**
     * @brief 发送断开连接的事件
     * @date 2018-10-30 21:57
     */
    bool disconnect()
    {
        string id = lmdb_signal<signal_type>::signal_key(_instanceName, DISCONNECT_END);
        bool bOK = lmdb_signal<signal_type>::signal_notify(id.c_str());
        if(bOK) _instanceMutex = nullptr;
        return bOK;
    }

    inline lmdb_event_connection& set(lmdb_env& wenv, const char* wkey = NULL)
    {
        _wenv = wenv;
        _sendkey = _createEvent(_wevent, wkey);
        return (*this);
    }
    inline lmdb_event_connection& set(lmdb_const_env& renv, const char* rkey = NULL)
    {
        _renv = renv;
        _recvkey = _createEvent(_revent, rkey);
        return (*this);
    }
    inline lmdb_event_connection& set(const char* rdb, const char* rkey = NULL)
    {
        _rdb = _strput(rdb);
        _recvkey = _createEvent(_revent, rkey);
        return (*this);
    }

    /**
     * @brief 删除读取事件的信号
     * @date 2018-11-03 14:55
     */
    bool remove_read()
    {
        if(!_revent.is_valid() || _recvkey.length() < 1)
            return false;
        if(!signal_type::remove(_recvkey.c_str()))
            return false;
        _recvkey = "";
        _revent = signal_type();
        return true;
    }
    /**
     * @brief 删除写入事件的信号
     * @date 2018-11-03 14:56
     */
    bool remove_write()
    {
        if(!_wevent.is_valid() || _sendkey.length() < 1)
            return false;
        if(!signal_type::remove(_sendkey.c_str()))
            return false;
        _sendkey = "";
        _wevent = signal_type();
        return true;
    }
    /**
     * @brief 清理所占用的系统资源
     * @date 2018-11-24 19:20:36
     */
    void cleanup()
    {
        dispose();
        remove_read();
        remove_write();
    }

    /// 克隆数据(用于子类迭代)
    void clone(const lmdb_event_connection& connection)
    {
        _wevent = connection._wevent;
        _wenv = connection._wenv;
        _sendkey = connection._sendkey;
        _revent = connection._revent;
        _recvkey = connection._recvkey;
        _rdb = connection._rdb;
        _renv = connection._renv;
        _instanceId = connection._instanceId;
        _instanceName = connection._instanceName;
        _instanceMutex = connection._instanceMutex;
    }
};
//--------------------------------------------------------- 
/**
 * @brief 异步的lmdb事件数据代理服务
 * @date 2018-10-12 16:05:55
 */
template<class TSignalType, class TLmdbValueConverter>
class lmdb_event_agent : public lmdb_event_connection<TSignalType>
{
public:
    typedef TSignalType signal_type;
    typedef TLmdbValueConverter value_converter;
    typedef lmdb_signal<signal_type> event_type;
    typedef lmdb_event_agent<signal_type, value_converter> self;
    typedef lmdb_event_connection<TSignalType> connection_type;
    typedef typename value_converter::value_type value_type;
    typedef lmdb_cursor_iterator<lmdb_string_converter, value_converter> data_iterator;
    typedef lmdb_cursor_const_iterator<lmdb_string_converter, value_converter> data_const_iterator;
public:
    lmdb_event_agent() {}
    lmdb_event_agent(const char* id) { create(id); }
    lmdb_event_agent(const char* id, lmdb_env& wenv, const char* wkey = NULL) { create(id);set(wenv, wkey); }
    lmdb_event_agent(const char* id, const char* rdb, const char* rkey = NULL) { create(id);set(rdb, rkey); }
    lmdb_event_agent(const char* id, lmdb_const_env& renv, const char* rkey = NULL) { create(id);set(renv, rkey); }
    lmdb_event_agent(const connection_type& connection) { clone(connection); }

    bool write(const value_type& data)
    {
        return event_type::write<value_converter>(_wevent, _wenv, _sendkey, data);
    }
    bool read(value_type& data, uint timeoutMs = 0)
    {
        return event_type::read<value_converter>(_revent, _renv, _recvkey, val, timeoutMs);
    }
    bool wait(value_type& data, uint timeoutMs, Ref<IInterrupter> interrupter = Ref<IInterrupter>())
    {
        int err = event_type::timed_wait<value_converter>(_revent, _rdb.c_str(), _recvkey, data, timeoutMs, interrupter);
        return err == DeviceError::Success;
    }
};
//--------------------------------------------------------- 
/**
 * @brief lmdb_connection断开连接的监听器
 * @date 2018-10-28 18:26:34
 */
template<class T>
class lmdb_disconnect_listener
{
public:
    typedef T signal_type;
    typedef lmdb_disconnect_listener<T> self;
    typedef std::function<void(const string&)> disconnect_callback_type;
    typedef lmdb_event_connection<signal_type> connection_type;
protected:
    bool _isRunning;
    list_t<std::pair<string, signal_type> > _queue;
    std::mutex _mutex;
    std::condition_variable_any _condition;
    std::shared_ptr<std::thread> _worker;
    std::vector<disconnect_callback_type> _disconnectCallbacks;

    string _getId(const string& name)
    {
        return lmdb_signal<signal_type>::signal_key(name, DISCONNECT_END);
    }

    /// 检测断开连接的通知线程函数
    void _run_disconnect()
    {
        while(_isRunning)
        {
            {
                std::unique_lock<std::mutex> lock(_mutex);
                if(_queue.size() < 1)
                {
                    _condition.wait(lock);
                    continue;
                }
                list_t<std::pair<string, signal_type> >::iterator itr;
                for(itr = _queue.begin();itr != _queue.end(); ++itr)
                {
                    if(itr->second.try_wait())
                    {
                        string name = itr->first;
                        string id = _getId(name);
                        signal_type::remove(id.c_str());
                        itr = list_helper<std::pair<string, signal_type> >::erase(_queue, itr);

                        std::vector<disconnect_callback_type>::iterator itr;
                        for(itr = _disconnectCallbacks.begin();itr != _disconnectCallbacks.end(); ++itr)
                        {
                            if((*itr) != nullptr)
                            {
                                (*itr)(name);
                            }
                        }
                    }
                }
            }
            Timer::Wait(DEV_OPERATOR_INTERVAL * 10);
        }
    }
public:
    lmdb_disconnect_listener() : _isRunning(false) {}
    virtual ~lmdb_disconnect_listener() { quit(); }

    inline void addCallback(disconnect_callback_type callback)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _disconnectCallbacks.push_back(callback);
    }
    inline void removeCallback(disconnect_callback_type callback)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        container_helper<std::vector<disconnect_callback_type> > >::remove(_disconnectCallbacks, callback);
    }
    void clearCallback()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _disconnectCallbacks.clear();
    }
    bool add(const connection_type& connection)
    {
        return add(connection.getName());
    }
    bool add(const string& name)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        string id = _getId(name);
        signal_type signal = signal_type::create_only(id.c_str());
        // 重复监听
        if(!signal.is_valid()) return false;
        _queue.push_back();
        _queue.back().first = name;
        _queue.back().second = signal;
        if(_worker == nullptr)
        {
            _isRunning = true;
            _worker = std::make_shared<std::thread>(std::bind(&self::_run_disconnect, this));
        }
        _condition.notify_all();
        return true;
    }
    bool remove(const string& name)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        list_t<std::pair<string, signal_type> >::iterator itr;
        for(itr = _queue.begin();itr != _queue.end(); ++itr)
        {
            if(itr->first == name)
            {
                string id = _getId(name);
                signal_type::remove(id.c_str());
                _queue.erase(itr);
                return true;
            }
        }
        return false;
    }
    void clear()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        list_t<std::pair<string, signal_type> >::iterator itr;
        for(itr = _queue.begin();itr != _queue.end(); ++itr)
        {
            string name = itr->first;
            string id = _getId(name);
            signal_type::remove(id.c_str());
        }
        _queue.clear();
    }
    inline bool is_running() const
    {
        return _worker != nullptr && _isRunning;
    }
    void quit()
    {
        if(!is_running()) return;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _isRunning = false;
            _condition.notify_all();
        }
        _worker->join();
        _worker = nullptr;

        clear();
    }
};
//--------------------------------------------------------- 
/**
 * @brief lmdb事件服务端
 *
 * @description 监听指定的事件,客户端将自己唯一的ID写到lmdb中,服务端根据ID创建与之对应的
 *
 * @date 2018-10-14 09:37:05
 */
template<class T>
class lmdb_event_server : public named_instance
{
public:
    typedef T signal_type;
    typedef lmdb_event_server<T> self;
    typedef lmdb_signal<signal_type> event_type;
    typedef lmdb_event_connection<signal_type> connection_type;
    typedef std::function<void(const string&)> disconnect_callback_type;
    typedef lmdb_cursor_iterator<lmdb_pod_converter<size_t>, lmdb_pod_converter<size_t> > lmdb_key_iterator;
    typedef lmdb_cursor_const_iterator<lmdb_pod_converter<size_t>, lmdb_pod_converter<size_t> > lmdb_key_const_iterator;

    /// 尝试连接(重复连接时将返回设备忙)
    static int try_connect(const char* svrkey, const char* dbRoot, connection_type& connection,
        uint timeoutMs = 0, Ref<IInterrupter> interrupter = Ref<IInterrupter>())
    {
        return connect(svrkey, dbRoot, connection, timeoutMs, interrupter, true);
    }
    /**
     * @brief 客户端连接
     * @date 2018-10-30 00:33
     * 
     * @param [in] svrkey 服务名称
     * @param [in] dbRoot lmdb存放目录
     * @param [out] connection 返回的连接
     * @param [in] timeoutMs [default:DEV_WAIT_TIMEOUT] 等待全局锁和连接的超时时间
     * @param [in] interrupter [default:null] 中断器
     * @param [in] isTryLock [default:false] 使用锁时是否尝试还是在超时时间内锁定
     */
    static int connect(const char* svrkey, const char* dbRoot, 
        connection_type& connection,
        uint timeoutMs = DEV_WAIT_TIMEOUT, 
        Ref<IInterrupter> interrupter = Ref<IInterrupter>(), 
        bool isTryLock = false)
    {
        string key = _strput(svrkey);
        // 保证当前访问server的实例只有一个
        boost::interprocess::named_mutex* pMutex = nullptr;
        try
        {
            string instanceName = named_instance::instanceKey(key);
            string mutexKey = named_instance::getHash(instanceName);
            pMutex = new boost::interprocess::named_mutex(boost::interprocess::open_only, mutexKey.c_str());
        }
        catch(const boost::interprocess::interprocess_exception&)
        {
            return DeviceError::DevNotExistErr;
        }
        std::shared_ptr<boost::interprocess::named_mutex> mutex(pMutex);
        if(isTryLock)
        {
            if(!mutex->try_lock()) return DeviceError::DevIsBusyErr;
        }
        else
        {
            int errWait = mutex_signal::timed_wait(*mutex, timeoutMs, interrupter);
            if(errWait != DeviceError::Success) return errWait;
        }
        string connectKey = connectId(key);
        Timer timer;
        while(timer.Elapsed() < timeoutMs)
        {
            if(event_type::signal_notify(connectKey.c_str()))
            {
                string acceptKey = acceptId(key);
                int err = event_type::signal_wait(acceptKey.c_str(), timeoutMs, interrupter);
                if(err == DeviceError::Success)
                {
                    bool bOK = _connect(key, _strput(dbRoot), connection);
                    err = bOK ? DeviceError::Success : DeviceError::OperatorErr;
                }
                mutex->unlock();
                return err;
            }
            /* 中断支持 */
            if(!interrupter.IsNull() && InterruptBehavior::Implement(*interrupter))
            {
                mutex->unlock();
                return DeviceError::OperatorInterruptErr;
            }
            Timer::Wait(DEV_OPERATOR_INTERVAL);
        }
        mutex->unlock();
        return DeviceError::WaitTimeOutErr;
    }
    /**
     * @brief 强行清理指定的connectionId
     * @date 2018-11-03 15:23
     */
    static void cleanup(const char* svrkey, size_t connectionId)
    {
        string instanceName = _strput(svrkey);
        string id = connectionKey(instanceName, connectionId);

        string sId = id;
        sId += ".";
        sId += REMOTE_END;
        event_type::remove(sId);
        named_instance::remove(sId);

        string rId = id;
        rId += ".";
        rId += LOCAL_END;
        event_type::remove(rId);
        named_instance::remove(rId);
    }

    /// 返回名称中的ID标识
    static size_t nameId(const string& name)
    {
        ByteArray n(name.c_str(), name.length());
        ByteBuilder buff(4);
        buff = StringConvert::Middle(n, '_', '.');
        size_t id = ArgConvert::FromString<size_t>(buff.GetString());
        return id;
    }
    /// 返回服务端的名称
    static string serviceName(const string& name)
    {
        ByteArray n(name.c_str(), name.length());
        if(!StringConvert::EndWith(n, REMOTE_END))
            return name;
        ByteBuilder itemName(n);
        itemName.RemoveTail(4);
        itemName += LOCAL_END;
        return string(itemName.GetString(), itemName.GetLength());
    }
    /// 返回客户端的名称
    static string clientName(const string& name)
    {
        ByteArray n(name.c_str(), name.length());
        if(!StringConvert::EndWith(n, LOCAL_END))
            return name;
        ByteBuilder itemName(n);
        itemName.RemoveTail(4);
        itemName += REMOTE_END;
        return string(itemName.GetString(), itemName.GetLength());
    }
    /// 返回accept事件的ID
    static string acceptId(const string& svrkey)
    {
        return named_instance::getHash(svrkey + ".accept");
    }
    /// 返回connect事件的ID
    static string connectId(const string& svrkey)
    {
        return named_instance::getHash(svrkey + ".connect");
    }
    /// 返回对应connectionId的连接Key
    static string connectionKey(const string& svrkey, size_t id)
    {
        string key = svrkey;
        key += "_";
        key += ArgConvert::ToString(id);
        return key;
    }
    /// 返回对应connectionId的本地ID
    static string connectionKeyID(const string& svrkey, const string& extendKey, size_t connectionId)
    {
        string id = connectionKey(svrkey, connectionId);
        id += extendKey;
        string sHashId = named_instance::getHash(id);
        return sHashId;
    }
    /// 返回服务对应的lmdb数据库目录
    static string svrDir(const string& svrkey, const string& dbRoot)
    {
        string dir = dbRoot;
        StringHelper::FormatDir(dir);
        dir += svrkey;
        return dir;
    }
protected:
    static bool _create_connection(const string& key, const string& dir, 
        const string& send, const string& recv, connection_type& connection)
    {
        string sendkey = key;
        sendkey += ".";
        sendkey += send;

        connection.create(sendkey.c_str());
        if(!connection.is_valid()) return false;

        string recvkey = key;
        recvkey += ".";
        recvkey += recv;

        string senddb = dir;
        StringHelper::FormatDir(senddb);
        senddb += send;

        string recvdb = dir;
        StringHelper::FormatDir(recvdb);
        recvdb += recv;

        lmdb_env wenv = lmdb_helper::open(senddb.c_str());
        lmdb_const_env renv = lmdb_helper::open(recvdb.c_str());
        if(!wenv.is_valid() || !renv.is_valid()) return false;
        connection.set(wenv, sendkey.c_str()).set(recvdb.c_str(), recvkey.c_str());
        bool bOK = connection.is_can_read() && connection.is_can_write();
        if(!bOK) connection.dispose();
        return bOK;
    }
    static bool _connect(const string& svrkey, const string& dbRoot, connection_type& connection)
    {
        string dir = svrDir(svrkey, dbRoot);
        lmdb_const_env env = lmdb_helper::open_readonly(dir.c_str());
        lmdb_key_const_iterator itr = lmdb_helper::open_readonly(env);
        if(!itr.is_valid()) return false;
        if(!itr.first() && itr.Key != 0) return false;
        string key = connectionKey(svrkey, itr.Value);
        string subdir = svrDir(key, dir);
        return _create_connection(key, subdir, REMOTE_END, LOCAL_END, connection);
    }
protected:
    typedef std::pair<size_t, size_t> connection_node;

    string _keyAccept;
    string _keyConnect;

    lmdb_env _env;
    /// 当前已连接客户端的id
    std::vector<connection_node> _connectionId;
    /// 历史已经释放,空闲的id
    std::vector<size_t> _historyId;
    /// 连接时尝试后无法使用的id
    std::vector<size_t> _disableId;
    /// 客户端请求连接的事件
    signal_type _connnectEvent;

    std::mutex _mutex;
    lmdb_disconnect_listener<signal_type> _listener;
    disconnect_callback_type _localCallback;
    disconnect_callback_type _remoteCallback;

    lmdb_event_server(const lmdb_event_server&) {}
    lmdb_event_server& operator=(const lmdb_event_server&) {}

    void _disconnectCallback(const string& name)
    {
        event_type::remove(name);
        named_instance::remove(name);

        ByteArray n(name.c_str(), name.length());
        size_t id = nameId(name);
        std::vector<connection_node>::iterator itr = _findId(id);
        if(itr == _connectionId.end()) return;
        if(StringConvert::EndWith(n, REMOTE_END))
        {
            itr->second &= 0x0F;
            if(_remoteCallback != nullptr)
            {
                _remoteCallback(name);
            }
        }
        else if(StringConvert::EndWith(n, LOCAL_END))
        {
            itr->second &= 0x0F0;
            if(_localCallback != nullptr)
            {
                _localCallback(name);
            }
        }
        // 已经没有收发端使用该连接
        if(itr->second == 0x00)
        {
            _connectionId.erase(itr);
            _historyId.push_back(id);
        }
    }
    std::vector<connection_node>::iterator _findId(size_t id)
    {
        std::vector<std::pair<size_t, size_t> >::iterator itr;
        for(itr = _connectionId.begin();itr != _connectionId.end(); ++itr)
        {
            if(itr->first == id)
                return itr;
        }
        return _connectionId.end();
    }
    size_t _createId()
    {
        if(_historyId.size() > 0)
        {
            size_t id = _historyId.back();
            _historyId.pop_back();
            return id;
        }
        
        std::vector<connection_node>::const_iterator itr;
        size_t index = 0;
        while(++index)
        {
            if(_findId(index) == _connectionId.end() &&
                !container_helper<std::vector<size_t> >::contains(_disableId, index))
            {
                break;
            }
        }
        return index;
    }
public:
    lmdb_event_server() : IsListenLocalDisconnectEvent(true) {}
    lmdb_event_server(const char* svrkey, const char* dbRoot) : IsListenLocalDisconnectEvent(true) { startup(svrkey, dbRoot); }
    virtual ~lmdb_event_server()
    {
        teardown();
    }

    /**
     * @brief 打开服务
     * @date 2018-10-20 17:53:13
     *
     * @param [in] svrkey 服务名称
     * @param [in] dbRoot 数据库目录
     * @param [in] isReload [default:false] 已存在服务时是否重新载入
     *
     * @retval DevAlreadyOpenErr : 服务已经存在
     * @retval DevConnectErr : db打开失败
     * @retval DevInvalidErr : db复位失败
     * @retval DevInitErr : db初始化失败
     *
     * @retval Success : 操作成功
     */
    int startup(const char* svrkey, const char* dbRoot, bool isReload = false)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        string name = _strput(svrkey);
        if(named_instance::is_exist(name))
        {
            if(!isReload)
                return DeviceError::DevAlreadyOpenErr;
            named_instance::remove(name);
        }
        string dir = svrDir(name, _strput(dbRoot));
        lmdb_env env = lmdb_helper::open(dir.c_str());
        lmdb_key_iterator itr = lmdb_helper::open(env);
        if(!itr.is_valid()) return DeviceError::DevConnectErr;
        if(!itr.put_commit(0, 0)) return DeviceError::DevInitErr;
        _env = env;
        create(svrkey);
        _keyConnect = connectId(_instanceName);
        _keyAccept = acceptId(_instanceName);
        _connnectEvent = signal_type::open_or_create(_keyConnect.c_str());
        _listener.addCallback(std::bind(&self::_disconnectCallback, this, std::placeholders::_1));
        return DeviceError::Success;
    }
    /// 关闭服务
    void teardown()
    {
        if(!is_valid()) return;
        std::unique_lock<std::mutex> lock(_mutex);
        lmdb_cursor cursor = lmdb_helper::open(_env);
        cursor.get_dbi().drop();
        std::vector<connection_node>::const_iterator itr;
        for(itr = _connectionId.begin();itr != _connectionId.end(); ++itr)
        {
            string id = connectionKey(_instanceName, itr->first);
            if(itr->second & 0x0F0)
            {
                string sId = id;
                sId += ".";
                sId += REMOTE_END;
                event_type::remove(sId);
                named_instance::remove(sId);
            }
            if(itr->second & 0x0F)
            {
                string rId = id;
                rId += ".";
                rId += LOCAL_END;
                event_type::remove(rId);
                named_instance::remove(rId);
            }
        }
        _disableId.clear();
        _historyId.clear();
        _connectionId.clear();
        _listener.quit();
        dispose();
        _env.close();
        signal_type::remove(_keyConnect.c_str());
        signal_type::remove(_keyAccept.c_str());
    }
    /**
     * @brief 监听客户端连接请求
     * @date 2018-10-14 10:03:21
     */
    int accept(connection_type& connection, uint timeoutMs = DEV_WAIT_TIMEOUT, 
        Ref<IInterrupter> interrupter = Ref<IInterrupter>())
    {
        std::unique_lock<std::mutex> lock(_mutex, std::defer_lock);
        if(!lock.try_lock()) return DeviceError::DevIsBusyErr;
        int err = event_type::signal_wait(_connnectEvent, timeoutMs, interrupter);
        if(err != DeviceError::Success)
            return err;
        lmdb_key_iterator itr = lmdb_helper::open(_env);
        // 生成connectionId时冲突最大尝试次数
        const size_t MAX_RETRY_COUNT = 0x0FF;
        size_t retryCount = 0;
        string key = "";
        size_t connectionId = _createId();
        for(size_t retryCount = 0; retryCount < MAX_RETRY_COUNT; ++retryCount)
        {
            if(!itr.is_valid() || !itr.put_commit(0, connectionId))
                return DeviceError::OperatorStatusErr;
            key = connectionKey(_instanceName, connectionId);
            string subdir = svrDir(key, _env.get_dir());
            if(_create_connection(key, subdir, LOCAL_END, REMOTE_END, connection))
                break;
            _disableId.push_back(connectionId);
            container_helper<std::vector<size_t> >::remove(_historyId, connectionId);

            /* 中断支持 */
            if(!interrupter.IsNull() && InterruptBehavior::Implement(*interrupter))
                return DeviceError::OperatorInterruptErr;

            if(!itr.revert())
                return DeviceError::OperatorStatusErr;
            connectionId = _createId();
        }
        if(retryCount >= MAX_RETRY_COUNT)
            return DeviceError::OperatorErr;
        Timer timer;
        while(timer.Elapsed() < timeoutMs)
        {
            if(event_type::signal_notify(_keyAccept.c_str()))
            {
                // 监听客户端断开连接的事件
                string remoteName = key;
                remoteName += ".";
                remoteName += REMOTE_END;
                _listener.add(remoteName);
                if(IsListenLocalDisconnectEvent) _listener.add(connection);
                _connectionId.push_back(connection_node(connectionId, 0x0FF));
                return DeviceError::Success;
            }
            /* 中断支持 */
            if(!interrupter.IsNull() && InterruptBehavior::Implement(*interrupter))
                return DeviceError::OperatorInterruptErr;
            Timer::Wait(DEV_OPERATOR_INTERVAL);
        }
        // 没有成功的产生通知,清理客户端连接的所有资源
        connection.cleanup();
        return DeviceError::RecvErr;
    }
    /// 返回对象是否有效
    inline bool is_valid() const
    {
        return _instanceMutex != nullptr && _env.is_valid();
    }
    /// 是否监听客户端断开连接的事件
    bool IsListenLocalDisconnectEvent;
    
    /// 设置服务端的断开事件回调
    disconnect_callback_type setLocalDisconnectCallback(disconnect_callback_type callback)
    {
        disconnect_callback_type last = _localCallback;
        _localCallback = callback;
        return last;
    }
    /// 设置客户端的断开事件回调
    disconnect_callback_type setRemoteDisconnectCallback(disconnect_callback_type callback)
    {
        disconnect_callback_type last = _remoteCallback;
        _remoteCallback = callback;
        return last;
    }
};
//--------------------------------------------------------- 
/**
 * @brief 用于管理lmdb_event_server生成的多个连接
 * @date 2018-11-19 22:59:27
 */
template<class TLmdbEventServer, class TLmdbConverter>
class lmdb_event_daemon : public TLmdbEventServer
{
public:
    typedef lmdb_event_daemon<TLmdbEventServer, TLmdbConverter> self;
    typedef typename TLmdbEventServer::signal_type signal_type;
    typedef TLmdbEventServer lmdb_server_type;
    typedef TLmdbConverter lmdb_converter_type;
    typedef lmdb_event_agent<signal_type, lmdb_converter_type> lmdb_agent_type;
    typedef std::function<void(std::shared_ptr<lmdb_agent_type>)> runable_type;
protected:
    std::shared_ptr<lmdb_agent_type> _cache;
    list_t<std::shared_ptr<std::thread> > _workers;
    list_t<std::shared_ptr<lmdb_agent_type> > _agents;
    typename lmdb_server_type::disconnect_callback_type _realRemoteCallback;
    /// 客户端连接断开时,自动断开服务端的连接
    void disconnect_local_callback(const string& name)
    {
        if(_realRemoteCallback != nullptr)
            _realRemoteCallback(name);

        string remoteName = lmdb_server_type::serviceName(name);
        ByteArray itemName(remoteName.c_str(), remoteName.length());

        list_t<std::shared_ptr<lmdb_agent_type> >::iterator itr;
        for(itr = _agents.begin();itr != _agents.end(); ++itr)
        {
            ByteArray n((*itr)->getName().c_str(), (*itr)->getName().length());
            if(StringConvert::Compare(n, itemName))
            {
                (*itr)->disconnect();
                _agents.erase(itr);
                return;
            }
        }
    }
public:
    /**
     * @brief 持续接受客户端连接(会阻塞当前线程)
     * @date 2018-11-19 23:18:25
     * 
     * @param [in] runable [default:null] 客户端连接后的线程函数
     * @param [in] timeoutMs [default:DEV_WAIT_TIMEOUT] 等待每个客户端的超时时间
     * @param [in] interrupter [default:null] 中断器
     */
    void running(runable_type runable = nullptr,
        uint timeoutMs = DEV_WAIT_TIMEOUT,
        Ref<IInterrupter> interrupter = Ref<IInterrupter>())
    {
        if(_cache == nullptr)
        {
            _cache = std::make_shared<lmdb_agent_type>();
        }
        _realRemoteCallback = TLmdbEventServer::_remoteCallback;
        TLmdbEventServer::_remoteCallback = std::bind(&self::disconnect_local_callback, this, std::placeholders::_1);
        while(true)
        {
            if(InterruptBehavior::Implement(interrupter))
                return;

            TLmdbEventServer::IsListenLocalDisconnectEvent = false;
            int err = TLmdbEventServer::accept(*_cache, timeoutMs, interrupter);
            if(err != DeviceError::Success)
                continue;

            std::unique_lock<std::mutex> lock(TLmdbEventServer::_mutex);
            _agents.push_back(_cache);
            _cache = std::make_shared<lmdb_agent_type>();
            if(runable == nullptr)
                continue;

            std::shared_ptr<std::thread> proc = std::make_shared<std::thread>(std::bind(runable, _agents.back()));
            _workers.push_back(proc);
        }
    }
    /**
     * @brief 等待所有线程执行完
     * @date 2018-11-19 23:33:02
     */
    void join_all()
    {
        list_t<std::shared_ptr<std::thread> >::iterator itr;
        for(itr = _workers.begin();itr != _workers.end(); ++itr)
        {
            if((*itr) != nullptr)
            {
                (*itr)->join();
            }
        }
        _workers.clear();
    }
    /**
     * @brief 服务端主动断开连接
     * @date 2018-11-24 12:13:44
     */
    bool disconnect(const string& name)
    {
        list_t<std::shared_ptr<lmdb_agent_type> >::iterator itr;
        for(itr = _agents.begin();itr != _agents.end(); ++itr)
        {
            if((*itr)->getName() == name)
            {
                bool bOK = (*itr)->disconnect();
                TLmdbEventServer::_disconnectCallback((*itr)->getName());
                _agents.erase(itr);
                return bOK;
            }
        }
        return false;
    }
};
//--------------------------------------------------------- 
} // namespace lmdb_extractor 
} // namespace wrapper
} // namespace zhou_yb 
//--------------------------------------------------------- 
#endif // _LIBZHOUYB_LMDB_EVENT_H_
//========================================================= 